package com.zxg.util.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConcurrentCollectionProcesser<T, RES> {
  private final static Logger             log                     = LoggerFactory.getLogger(ConcurrentCollectionProcesser.class);

  public static final Integer             DEFAULT_CONCURRENT_SIZE = 4;
  public static final Integer             MAX_CONCURRENT_SIZE     = 16;
  public static final Integer             EVERY_TASK_ITEM_SIZE    = 100;
  private List<T>                         objects;
  private Integer                         concurrentSize          = DEFAULT_CONCURRENT_SIZE;
  private FutureTaskPool<List<RES>>       taskThreadPool          = null;
  private ConcurrentSigleExecutor<T, RES> executor;

  private List<FutureTask<List<RES>>>     futureTasks;

  public ConcurrentCollectionProcesser(List<T> objects, ConcurrentSigleExecutor<T, RES> executor) {
    this(objects, DEFAULT_CONCURRENT_SIZE, executor);
  }

  public ConcurrentCollectionProcesser(Integer concurrentSize, ConcurrentSigleExecutor<T, RES> executor) {
    super();
    if (concurrentSize <= MAX_CONCURRENT_SIZE && concurrentSize > 0)
      this.concurrentSize = concurrentSize;
    else
      this.concurrentSize = DEFAULT_CONCURRENT_SIZE;
    taskThreadPool = new FutureTaskPool<List<RES>>(concurrentSize);
    this.executor = executor;
  }

  public ConcurrentCollectionProcesser(List<T> objects, Integer concurrentSize, ConcurrentSigleExecutor<T, RES> executor) {
    super();
    if (concurrentSize <= MAX_CONCURRENT_SIZE && concurrentSize > 0)
      this.concurrentSize = concurrentSize;
    else
      this.concurrentSize = DEFAULT_CONCURRENT_SIZE;
    taskThreadPool = new FutureTaskPool<List<RES>>(concurrentSize);
    this.executor = executor;

    updateObjects(objects);

  }

  public void updateObjects(List<T> objects) {
    if (objects != null && objects.size() >= 0) {
      this.objects = objects;

      if (objects.size() < concurrentSize)
        this.concurrentSize = objects.size();
      if (this.concurrentSize <= 0)
        this.concurrentSize = DEFAULT_CONCURRENT_SIZE;
      futureTasks = new ArrayList<FutureTask<List<RES>>>();
    }

  }

  public List<RES> exec(List<T> objects) {
    updateObjects(objects);
    return this.exec();
  }

  public List<RES> exec() {
    if (this.objects == null || this.objects.size() == 0) {
      return null;
    }
    int totalSize = this.objects.size();
    int everySize = totalSize / this.concurrentSize;
    int count = 0;
    if (everySize > EVERY_TASK_ITEM_SIZE) {
      everySize = EVERY_TASK_ITEM_SIZE;
      count = totalSize / everySize;
      if (totalSize % everySize != 0)
        count = count + 1;
    } else if (everySize == 0) {
      count = totalSize;
      everySize = 1;
    } else {
      count = this.concurrentSize;
    }

    for (int i = 0; i < count; i++) {
      List<T> list = new ArrayList<T>();
      int start = i * everySize;
      int end = 0;
      if (i == count - 1) {
        end = this.objects.size();
      } else {
        end = start + everySize;
      }
      if (end > this.objects.size())
        end = this.objects.size();
      if (start > end)
        break;
      list.addAll(this.objects.subList(start, end));
      futureTasks.add(taskThreadPool.submit(new ConcurrentSigleTask(list)));
    }

    List<RES> result = new ArrayList<RES>();
    for (FutureTask<List<RES>> future : futureTasks) {
      try {
        List<RES> res = future.get();
        if (res != null) {
          result.addAll(res);
        }
      } catch (InterruptedException e) {
        log.error("error:", e);
        continue;
      } catch (ExecutionException e) {
        log.error("error:", e);
        continue;
      }
    }

    return result;
  }

  class ConcurrentSigleTask implements Callable<List<RES>> {

    private List<T> subObjects;

    public ConcurrentSigleTask(List<T> subObjects) {
      super();
      this.subObjects = subObjects;
    }

    @Override
    public List<RES> call() throws Exception {
      if (subObjects == null || subObjects.size() == 0)
        return new ArrayList<RES>();
      List<RES> res = new ArrayList<RES>();
      for (T obj : subObjects) {
        RES o = executor.process(obj);
        if (o != null) {
          res.add(o);
        }
      }
      return res;
    }

  }

  public static void main(String[] args) {
    List<String> param = new ArrayList<String>();
    for (int i = 0; i < 3; i++) {
      param.add("p_" + i);
    }
    ConcurrentCollectionProcesser<String, String> process = new ConcurrentCollectionProcesser<String, String>(null, 10,
        new ConcurrentSigleExecutor<String, String>() {

          @Override
          public String process(String obj) {
            // TODO Auto-generated method stub
            return obj + "_" + System.currentTimeMillis();
          }
        });
    process.updateObjects(param);
    System.out.println(process.exec());
    param = new ArrayList<String>();
    for (int i = 100; i < 1000; i++) {
      param.add("p_" + i);
    }

    // process.updateObjects(param);
    System.out.println(process.exec(param));
  }
}
